/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2023-2023. All rights reserved.
 * Description: ffts plus task dispatcher
 * Author: lilianlin
 * Create: 2023-05-01
 */

#include "externalinput_pub.h"
#include "adapter_rts.h"
#include "log.h"
#include "dtype_common.h"
#include "dispatcher_ffts.h"

constexpr u32 UB_BLOCK_SIZE = 32;
constexpr u64 TBE_REDUCE_MAX_COUNT = INT32_MAX;

namespace hccl {
DispatcherFFTS::DispatcherFFTS(const s32 deviceLogicId)
    : DispatcherPub(deviceLogicId), fftsCtxsPtr(nullptr), disableFfts_(true)
{}

DispatcherFFTS::~DispatcherFFTS()
{}

void DispatcherFFTS::JudgeFftsCtxInitialized(bool &fftsCtxInitFlag)
{
    fftsCtxInitFlag = fftsCtxsPtr != nullptr;
}

void DispatcherFFTS::SetNormalMode()
{
    disableFfts_ = true;
}

HcclResult DispatcherFFTS::ResetFftsCtx(bool enableCache, const std::string &key)
{
    disableFfts_ = false;
    if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
        HCCL_DEBUG("ffts task is disabled.");
        disableFfts_ = true;
    } else {
        fftsCtxsPtr = fftsCtxProvider.GetFftsCtx(enableCache, key);
        CHK_PTR_NULL(fftsCtxsPtr);
        disableFfts_ = false;
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::LaunchTasksEx(Stream &stream, std::vector<Stream> &subStreams)
{
    if (UNLIKELY(disableFfts_)) {
        return HCCL_SUCCESS;
    }

    uint64_t beginTime = GetMsprofSysCycleTime();
    CHK_PTR_NULL(fftsCtxsPtr); // 检查Context是否进行过Reset
    
    if (!fftsCtxsPtr->completed) {
        fftsCtxsPtr->ctxNum = fftsCtxsPtr->refreshIndex;
        fftsCtxsPtr->completed = true;
    }

    if (fftsCtxsPtr->refreshIndex != fftsCtxsPtr->ctxNum) {
        HCCL_ERROR("ffts context num is invaild, expected:%u, actual:%u.", fftsCtxsPtr->ctxNum,
            fftsCtxsPtr->refreshIndex);
        return HCCL_E_PARA;
    }

    if (fftsCtxsPtr->ctxNum == 0) {
        HCCL_INFO("ffts context num is 0, will not submit this context.");
        return HCCL_SUCCESS;
    }

    if (fftsCtxsPtr->ctxNum > HCCL_FFTS_CAPACITY) {
        HCCL_ERROR("[DispatcherFFTS][LaunchTasksEx] CtxNum[%u] exceeds the limit of FFTS+ graph.", fftsCtxsPtr->ctxNum);
        return HCCL_E_NOT_SUPPORT;
    }

    fftsCtxsPtr->refreshIndex = 0;

    rtFftsPlusSqe_t fftsPlusSqe = {};
    ConstructFftsSqe(fftsPlusSqe);

    rtFftsPlusTaskInfo_t task = {};
    task.argsHandleInfoNum = 0;
    task.argsHandleInfoPtr = nullptr;
    ConstructFftsTask(task, fftsPlusSqe);

    PrintFFTSDebugDetails(fftsPlusSqe, task);
    CHK_RET(hrtFftsPlusTaskLaunchWithFlag(&task, stream.ptr(), 0));
    disableFfts_ = true;
    // 调用回调来保存task信息
    if (callback_ != nullptr) {
        struct TaskPara taskPara;
        taskPara.type = TaskType::TASK_GRAPH_LAUNCH;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.graphLaunch.ctxNum = fftsCtxsPtr->ctxNum;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }

    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::ConstructFftsSqe(rtFftsPlusSqe_t &fftsPlusSqe)
{
    fftsPlusSqe.fftsType = RT_FFTS_PLUS_TYPE;
    fftsPlusSqe.totalContextNum = static_cast<uint16_t>(fftsCtxsPtr->ctxNum);
    fftsPlusSqe.readyContextNum = 1;
    fftsPlusSqe.preloadContextNum = (fftsPlusSqe.totalContextNum <= CONTEXT_MAX_NUM ?\
        fftsPlusSqe.totalContextNum : CONTEXT_MAX_NUM);

    // 配置notify wait 超时时间
    // 因为老版本用户设置HCCL_EXEC_TIMEOUT为0，hccl将0传递给rts,rts将0转换为1770s传递给硬件, 未达到永不超时效果，不符合预期；
    // 所以现版用户配置为0时，hccl转换成65535，rts识别到65535后会又会转换成0，去硬件设置永不超时
    if (GetExternalInputHcclExecTimeOut() == 0) {
        fftsPlusSqe.timeout = FFTS_TIMEOUT_MAX;
    // 因为65535被当作永不超时处理，所以当用户配置65535时需要改变他的值，防止误错做成永不超时
    } else if (GetExternalInputHcclExecTimeOut() == FFTS_TIMEOUT_MAX) {
        fftsPlusSqe.timeout = FFTS_TIMEOUT_MAX - 1;
    } else {
        fftsPlusSqe.timeout = GetExternalInputHcclExecTimeOut();
    }
    // 标识通信task，优化调度性能 (RTS要求带AIV/AIC任务的是 0x5B, 否则是0x5A)
    fftsPlusSqe.subType = argsHandleList.empty() ? 0x5A : 0x5B;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::ConstructFftsTask(rtFftsPlusTaskInfo_t &task, rtFftsPlusSqe_t &fftsPlusSqe)
{
    task.fftsPlusSqe = &fftsPlusSqe;
    task.descBuf = fftsCtxsPtr->contexts.data();
    task.descBufLen = sizeof(rtFftsPlusComCtx_t) * fftsCtxsPtr->ctxNum;
    task.descAddrType = 0;
    if (!argsHandleList.empty()) {
        task.argsHandleInfoNum = argsHandleList.size();
        task.argsHandleInfoPtr = argsHandleList.data();
        argsHandleList.clear();
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::PrintFFTSDebugDetails(rtFftsPlusSqe_t &fftsPlusSqe, rtFftsPlusTaskInfo_t &task)
{
    if (HcclCheckLogLevel(HCCL_LOG_DEBUG)) {
        HCCL_DEBUG("-------------------------------");
        HCCL_DEBUG("totalContextNum:0x%04x", fftsPlusSqe.totalContextNum);
        HCCL_DEBUG("readyContextNum:0x%04x", fftsPlusSqe.readyContextNum);
        HCCL_DEBUG("preloadContextNum:0x%04x", fftsPlusSqe.preloadContextNum);
        HCCL_DEBUG("descBuf:%p", task.descBuf);
        HCCL_DEBUG("descBufLen:%u", task.descBufLen);
        HCCL_DEBUG("descAddrType:%u", task.descAddrType);
        const u32 printLineNum = 32;
        const u32 printBytePreLine = sizeof(rtFftsPlusComCtx_t) / printLineNum;
        const u32 byte3Offset = 3;
        const u32 byte2Offset = 2;
        const u32 byte1Offset = 1;
        const u32 byte0Offset = 0;
        for (u32 i = 0; i < fftsCtxsPtr->ctxNum; i++) {
            HCCL_DEBUG("-------------------------------");
            HCCL_DEBUG("index:0x%02x", i);
            for (u32 j = 0; j < printLineNum; j++) {
                u8 *prtPtr = reinterpret_cast<u8 *>(reinterpret_cast<u64>(task.descBuf) +
                    i * sizeof(rtFftsPlusComCtx_t) + j * printBytePreLine);
                HCCL_DEBUG("%02x %02x %02x %02x", *(prtPtr + byte3Offset), *(prtPtr + byte2Offset),
                    *(prtPtr + byte1Offset), *(prtPtr + byte0Offset));
            }
        }
        HCCL_DEBUG("-------------------------------");
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::UpdataFftsDescRelationship(Stream &stream)
{
    if (fftsCtxsPtr->refreshIndex > 0) {
        u32 lastThreadPreIndex = fftsCtxsPtr->lastThreadIndex[stream.id()];
        if (lastThreadPreIndex > 0) {
            fftsCtxsPtr->contexts[lastThreadPreIndex - 1]
                .successorList[fftsCtxsPtr->contexts[lastThreadPreIndex - 1].successorNum] = fftsCtxsPtr->refreshIndex;
            fftsCtxsPtr->contexts[lastThreadPreIndex - 1].successorNum++;
            fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex].predCntInit++;
            fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex].predCnt++;
        }
    }

    std::vector<u32> &notifyVec = fftsCtxsPtr->unassignedSuccessorEnd[stream.id()];

    for (std::vector<u32>::iterator iterNotify = notifyVec.begin(); iterNotify != notifyVec.end();) {
        std::vector<u32>::iterator iter = std::find(fftsCtxsPtr->ignoredSuccessorStart.begin(),
            fftsCtxsPtr->ignoredSuccessorStart.end(), *iterNotify);
        if (iter != fftsCtxsPtr->ignoredSuccessorStart.end()) {
            // signal record位于stream上的首个task，该signal无需构建successor连接。
            fftsCtxsPtr->ignoredSuccessorStart.erase(iter);
            iterNotify = notifyVec.erase(iterNotify);
        } else {
            if (fftsCtxsPtr->unrelatedSuccessorStart.find(*iterNotify) != fftsCtxsPtr->unrelatedSuccessorStart.end()) {
                u32 preIndex = fftsCtxsPtr->unrelatedSuccessorStart[*iterNotify];
                fftsCtxsPtr->contexts[preIndex].successorList[fftsCtxsPtr->contexts[preIndex].successorNum] =
                    fftsCtxsPtr->refreshIndex;
                fftsCtxsPtr->contexts[preIndex].successorNum++;
                fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex].predCntInit++;
                fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex].predCnt++;

                fftsCtxsPtr->unrelatedSuccessorStart.erase(*iterNotify);
                iterNotify = notifyVec.erase(iterNotify);
            } else {
                fftsCtxsPtr->unrelatedSuccessorEnd[*iterNotify] = fftsCtxsPtr->refreshIndex;
                iterNotify = notifyVec.erase(iterNotify);
            }
        }
    }
    return HCCL_SUCCESS;
}

void DispatcherFFTS::EnsureFftsContextsSize()
{
    if (fftsCtxsPtr->refreshIndex >= fftsCtxsPtr->contexts.size()) {
        fftsCtxsPtr->contexts.resize(fftsCtxsPtr->contexts.size() * 2);     // context空间不足，申请当前context num的2倍
    }
    return;
}

HcclResult DispatcherFFTS::InitFftsDescNotifyRecordRemote(HcclRtNotify signal, Stream &stream, u64 &signalAddr)
{
    EnsureFftsContextsSize();
    rtFftsPlusComCtx_t &comCtx = fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex];
    memset_s(&comCtx, sizeof(rtFftsPlusComCtx_t), 0, sizeof(rtFftsPlusComCtx_t));

    rtFftsPlusWriteValueCtx_t* ctx = reinterpret_cast<rtFftsPlusWriteValueCtx_t*>(&comCtx);

    CHK_RET(ConstructFftsNotifyRecordRemoteCtx(signal, ctx, signalAddr));

    CHK_RET(UpdataFftsDescRelationship(stream));

    fftsCtxsPtr->refreshIndex++;
    fftsCtxsPtr->lastThreadIndex[stream.id()] = fftsCtxsPtr->refreshIndex;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::ConstructFftsNotifyRecordRemoteCtx(HcclRtNotify &signal, rtFftsPlusWriteValueCtx_t *ctx,
    u64 &signalAddr)
{
    const u64 u64HighMask = 0xffffffff00000000;
    const u64 u64LowMask = 0x00000000ffffffff;
    const u32 shift = 32;
    u64 notifyAddr = signalAddr;
    if (notifyAddr == INVALID_U64) {
        CHK_RET(hrtNotifyGetAddr(signal, &notifyAddr));
    }
    ctx->contextType = RT_CTX_TYPE_WRITE_VALUE;
    ctx->successorNum = 0;
    ctx->aten = 0;
    ctx->predCntInit = 0;
    ctx->predCnt = 0;
    ctx->atm = 0;
    ctx->threadId = 0;
    ctx->threadDim = 1;
    ctx->awSize = 2; // 2: write 4 bytes
    ctx->awSnoop = 0;
    ctx->awCache = 0;
    ctx->awProt = 0;
    ctx->awVa = 0;  // 写物理地址（global 表单 write value 的va校验要关闭）
    ctx->res11 = 4; // 4: 标识notify record
    ctx->writeAddressBaseL = notifyAddr & u64LowMask;
    ctx->writeAddressBaseH = (notifyAddr & u64HighMask) >> shift;
    ctx->writeAddressOffset = 0;
    ctx->writeValue[0] = 1; // index 1: byte0~3
    ctx->writeValue[1] = 0; // index 1: byte4~7
    ctx->writeValue[2] = 0; // index 2: byte8~11
    ctx->writeValue[3] = 0; // index 3: byte12~15
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InitFftsDescNotifyWait(HcclRtNotify signal, Stream &stream)
{
    EnsureFftsContextsSize();
    rtFftsPlusComCtx_t &comCtx = fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex];
    memset_s(&comCtx, sizeof(rtFftsPlusComCtx_t), 0, sizeof(rtFftsPlusComCtx_t));

    rtFftsPlusNotifyCtx_t* ctx = reinterpret_cast<rtFftsPlusNotifyCtx_t*>(&comCtx);

    ConstructFFtsNotifyWaitCtx(signal, ctx);
    CHK_RET(UpdataFftsDescRelationship(stream));

    fftsCtxsPtr->refreshIndex++;
    fftsCtxsPtr->lastThreadIndex[stream.id()] = fftsCtxsPtr->refreshIndex;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::ConstructFFtsNotifyWaitCtx(HcclRtNotify &signal, rtFftsPlusNotifyCtx_t* ctx)
{
    u32 notifyID = 0;
    CHK_RET(hrtGetNotifyID(signal, &notifyID));
    return ConstructFftsNotifyCtx(notifyID, RT_CTX_TYPE_NOTIFY_WAIT, ctx);
}

HcclResult DispatcherFFTS::ConstructFftsNotifyCtx(u32 notifyID, uint16_t contextType, rtFftsPlusNotifyCtx_t* ctx)
{
    ctx->contextType = contextType;
    ctx->successorNum = 0;
    ctx->aten = 0;
    ctx->predCntInit = 0;
    ctx->predCnt = 0;
    ctx->satm = 0;
    ctx->atm = 0;
    ctx->threadId = 0;
    ctx->threadDim = 1;
    ctx->notifyIdBase = notifyID;
    ctx->autoWindow = 0;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InitFftsDescSdma(void *dst, const void *src, u64 cnt, Stream &stream, u32 sdmaSqeHeader)
{
    uint64_t spiltLoop = 0;
    uint64_t addrOffset = 0;
    uint64_t contSplit = 0;
    if (cnt > HCCL_SDMA_MAX_COUNT_4GB) {
        spiltLoop = (cnt % HCCL_SDMA_MAX_COUNT_4GB) ?
            (cnt / HCCL_SDMA_MAX_COUNT_4GB) : (cnt / HCCL_SDMA_MAX_COUNT_4GB - 1);
        HCCL_INFO("ffts+ InitFftsDescSdma SDMA task countSize is bigger than 4GB and do segmentation splitloop[%llu]",
            spiltLoop);
    }
    /* ffts+ SDMA任务拆分处理 */
    for (uint64_t index = 0 ; index <= spiltLoop; index++) {
        addrOffset = index * HCCL_SDMA_MAX_COUNT_4GB;
        contSplit = (index == spiltLoop) ? (cnt - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
        const void *srcSplit = (const void*)(static_cast<char *>(const_cast<void*>(src)) + addrOffset);
        void *dstSplit = static_cast<void *>(static_cast<char *>(dst) + addrOffset);

        EnsureFftsContextsSize();
        rtFftsPlusComCtx_t &comCtx = fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex];
        memset_s(&comCtx, sizeof(rtFftsPlusComCtx_t), 0, sizeof(rtFftsPlusComCtx_t));

        rtFftsPlusSdmaCtx_t* ctx = reinterpret_cast<rtFftsPlusSdmaCtx_t*>(&comCtx);

        ConstructFftsSdmaCtx(dstSplit, srcSplit, contSplit, sdmaSqeHeader, ctx, true);
        CHK_RET(UpdataFftsDescRelationship(stream));

        fftsCtxsPtr->refreshIndex++;
        fftsCtxsPtr->lastThreadIndex[stream.id()] = fftsCtxsPtr->refreshIndex;
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::ConstructFftsSdmaCtx(void *dst, const void *src, u64 cnt, u32 sdmaSqeHeader,
    rtFftsPlusSdmaCtx_t *ctx, bool isInit)
{
    if (isInit) {
        ctx->successorNum = 0;
        ctx->aten = 0;
        ctx->predCntInit = 0;
        ctx->predCnt = 0;
        ctx->atm = 0;
        ctx->pmg = 0;
        ctx->ns = 0;
        ctx->partId = 0;
        ctx->qos = 0;
        ctx->threadId = 0;
        ctx->threadDim = 1;

        ctx->sourceStreamId = 0;
        ctx->sourceSubstreamId = 0;
        ctx->sourceAddressOffset = 0;
        ctx->destinationStreamId = 0;
        ctx->destinationSubstreamId = 0;
        ctx->destinationAddressOffset = 0;
    }

    if (qosCfg_ != INVALID_QOSCFG) {
        DevType deviceType;
        CHK_RET(hrtGetDeviceType(deviceType));
        // 910B、910_93 qos值芯片约束, 默认是6
        if (deviceType != DevType::DEV_TYPE_910B || deviceType != DevType::DEV_TYPE_910_93) {
            ctx->qos = QOS_DEFAULT_VALUE;
        } else {
            ctx->qos = qosCfg_ & 0xFU;
            ctx->partId = (qosCfg_ >> 4U) & 0xFFU;
        }
        HCCL_DEBUG("qosCfg[%x] cfgInfo.qos[%u] cfgInfo.partId[%u]", qosCfg_, ctx->qos, ctx->partId);
    }

    ctx->contextType = (cnt == 0 || src == dst) ? RT_CTX_TYPE_LABEL : RT_CTX_TYPE_SDMA;
    ctx->res3 = 0x5A;  // 0x5A tell mcu that it is an HCCL label ctx.
    ctx->sdmaSqeHeader = sdmaSqeHeader;

    const u64 u64HighMask = 0xffffffff00000000;
    const u64 u64LowMask = 0x00000000ffffffff;
    const u32 shift = 32;

    ctx->sourceAddressBaseL = reinterpret_cast<u64>(src) & u64LowMask;
    ctx->sourceAddressBaseH = (reinterpret_cast<u64>(src) & u64HighMask) >> shift;

    ctx->destinationAddressBaseL = reinterpret_cast<u64>(dst) & u64LowMask;
    ctx->destinationAddressBaseH = (reinterpret_cast<u64>(dst) & u64HighMask) >> shift;

    ctx->nonTailDataLength = cnt;
    ctx->tailDataLength = cnt;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::RefreshFftsDescSdma(void *dst, const void *src, u64 cnt, Stream &stream, u32 sdmaSqeHeader)
{
    rtFftsPlusComCtx_t &comCtx = fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex];
    rtFftsPlusSdmaCtx_t* ctx = reinterpret_cast<rtFftsPlusSdmaCtx_t*>(&comCtx);

    if ((ctx->contextType != RT_CTX_TYPE_SDMA) && (ctx->contextType != RT_CTX_TYPE_LABEL)) {
        HCCL_ERROR("ffts context type is invalid, expected:0x%x, actual:0x%x", ctx->contextType, RT_CTX_TYPE_SDMA);
        return HCCL_E_PARA;
    }

    ConstructFftsSdmaCtx(dst, src, cnt, sdmaSqeHeader, ctx, false);

    fftsCtxsPtr->refreshIndex++;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InitFftsDescRdmaSend(u32 dbindex, u64 dbinfo, Stream &stream)
{
    EnsureFftsContextsSize();
    rtFftsPlusComCtx_t &comCtx = fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex];
    memset_s(&comCtx, sizeof(rtFftsPlusComCtx_t), 0, sizeof(rtFftsPlusComCtx_t));

    rtFftsPlusWriteValueCtx_t* ctx = reinterpret_cast<rtFftsPlusWriteValueCtx_t*>(&comCtx);

    ConstructFftsWriteValueCtx(dbindex, dbinfo, ctx);
    CHK_RET(UpdataFftsDescRelationship(stream));

    fftsCtxsPtr->refreshIndex++;
    fftsCtxsPtr->lastThreadIndex[stream.id()] = fftsCtxsPtr->refreshIndex;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::ConstructFftsWriteValueCtx(u32 dbindex, u64 dbinfo, rtFftsPlusWriteValueCtx_t *ctx)
{
    const u64 u64HighMask = 0xffffffff00000000;
    const u64 u64LowMask = 0x00000000ffffffff;
    const u32 shift = 32;
    u64 dbAddr = 0;

    if (!IsInvalidRdmaParam(dbindex, dbinfo)) {
        CHK_RET(hrtGetRdmaDoorbellAddr(dbindex, dbAddr));
    }

    ctx->contextType =  IsInvalidRdmaParam(dbindex, dbinfo) ? RT_CTX_TYPE_LABEL : RT_CTX_TYPE_WRITE_VALUE;
    ctx->successorNum = 0;
    ctx->aten = 0;
    ctx->predCntInit = 0;
    ctx->predCnt = 0;
    ctx->atm = 0;
    ctx->threadId = 0;
    ctx->threadDim = 1;
    ctx->awSize = 3; // 3: write 8 Bytes
    ctx->awSnoop = 0;
    ctx->awCache = 0;
    ctx->awProt = 0;
    ctx->awVa = 0;  // 写物理地址（global 表单 write value 的va校验要关闭）
    ctx->res11 = 2; // 2: 标识rdma send
    ctx->writeAddressBaseL = dbAddr & u64LowMask;
    ctx->writeAddressBaseH = (dbAddr & u64HighMask) >> shift;
    ctx->writeAddressOffset = 0;
    ctx->writeValue[0] = dbinfo & u64LowMask;             // index 0: byte0~3
    ctx->writeValue[1] = (dbinfo & u64HighMask) >> shift; // index 1: byte4~8
    ctx->writeValue[2] = 0;                               // index 2: byte8~11
    ctx->writeValue[3] = 0;                               // index 3: byte12~15
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::RefreshFftsDescRdmaSend(u32 dbindex, u64 dbinfo, Stream &stream)
{
    const u64 u64HighMask = 0xffffffff00000000;
    const u64 u64LowMask = 0x00000000ffffffff;
    const u32 shift = 32;
    u64 dbAddr = 0;

    rtFftsPlusComCtx_t &comCtx = fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex];
    rtFftsPlusWriteValueCtx_t *ctx = reinterpret_cast<rtFftsPlusWriteValueCtx_t *>(&comCtx);

    if (ctx->contextType != RT_CTX_TYPE_WRITE_VALUE && ctx->contextType != RT_CTX_TYPE_LABEL) {
        HCCL_ERROR("ffts context type is invaild, expected:0x%x, actual:0x%x or 0x%x", ctx->contextType,
            RT_CTX_TYPE_WRITE_VALUE, RT_CTX_TYPE_LABEL);
        return HCCL_E_PARA;
    };

    if (IsInvalidRdmaParam(dbindex, dbinfo)) {
        ctx->contextType = RT_CTX_TYPE_LABEL;
        ctx->writeAddressBaseL = 0;
        ctx->writeAddressBaseH = 0;
        SkipFftsRefresh();
        return HCCL_SUCCESS;
    }

    CHK_RET(hrtGetRdmaDoorbellAddr(dbindex, dbAddr));
    if (ctx->contextType == RT_CTX_TYPE_LABEL) {
        ctx->contextType = RT_CTX_TYPE_WRITE_VALUE;
        ctx->writeAddressBaseL = dbAddr & u64LowMask;
        ctx->writeAddressBaseH = (dbAddr & u64HighMask) >> shift;
    } else {
        u64 dbAddrOrg = (static_cast<u64>(ctx->writeAddressBaseH) << 32) + ctx->writeAddressBaseL;
        if (dbAddrOrg != dbAddr) {
            HCCL_ERROR("ffts context dbAddr is invaild, expected:0x%x, actual:0x%x", dbAddrOrg, dbAddr);
            return HCCL_E_PARA;
        }
    }

    ctx->writeValue[0] = dbinfo & u64LowMask;                           // index 0: byte0~3
    ctx->writeValue[1] = (dbinfo & u64HighMask) >> shift;               // index 1: byte4~8
    ctx->writeValue[2] = 0;                                             // index 2: byte8~11
    ctx->writeValue[3] = 0;                                             // index 3: byte12~15

    fftsCtxsPtr->refreshIndex++;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InitFftsDescMemcpy(DeviceMem &dst, const DeviceMem &src, Stream &stream)
{
    CHK_RET(InitFftsDescSdma(dst.ptr(), src.ptr(), src.size(), stream, SDMA_FP32_ATOMIC_MOVE_SQE));
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::RefreshFftsDescMemcpy(DeviceMem &dst, const DeviceMem &src, Stream &stream)
{
    CHK_RET(RefreshFftsDescSdma(dst.ptr(), src.ptr(), src.size(), stream, SDMA_FP32_ATOMIC_MOVE_SQE));
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::GetSdmaSqeHeader(const HcclDataType &datatype, const HcclReduceOp &redOp,
    FftsSdmaSqeHeader &sdmaSqeHeader)
{
    sdmaSqeHeader.word = 0;
    sdmaSqeHeader.bit.dns = 1;
    sdmaSqeHeader.bit.sns = 1;
    sdmaSqeHeader.bit.dssv = 1;
    sdmaSqeHeader.bit.sssv = 1;
    u32 SDMA_DTYPE_TABLE[HCCL_DATA_TYPE_RESERVED] = {
        0,      // int8
        1,      // int16
        2,      // int32
        6,      // fp16
        7,      // fp32
        0xf,    // int64 不支持
        0xf,    // uint64 不支持
        0xf,    // uint8 不支持
        0xf,    // uint16 不支持
        0xf,    // uint32 不支持
        0xf,    // fp64 不支持
        0x8     // bfp16
    };

    u32 SDMA_OPCODE_TABLE[HCCL_REDUCE_RESERVED] = {
        1,      // sum
        0xf,    // prod
        2,      // max
        3       // min
    };
    sdmaSqeHeader.bit.datatype = SDMA_DTYPE_TABLE[datatype];
    CHK_PRT_RET((sdmaSqeHeader.bit.datatype == 0xf),
        HCCL_ERROR("sdma not support dtype %s", GetDataTypeEnumStr(datatype).c_str()), HCCL_E_PARA);
    sdmaSqeHeader.bit.opcode = SDMA_OPCODE_TABLE[redOp];
    CHK_PRT_RET((sdmaSqeHeader.bit.opcode == 0xf),
        HCCL_ERROR("sdma not support reducetype %s", GetReduceOpEnumStr(redOp).c_str()), HCCL_E_PARA);
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InitFftsDescSdmaReduce(void *dst, const void *src, u64 cnt, const HcclDataType &datatype,
    const HcclReduceOp &redOp, Stream &stream)
{
    FftsSdmaSqeHeader sdmaSqeHeader;
    CHK_RET(GetSdmaSqeHeader(datatype, redOp, sdmaSqeHeader));
    CHK_RET(InitFftsDescSdma(dst, src, cnt * SIZE_TABLE[datatype], stream, sdmaSqeHeader.word));
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::RefreshFftsDescSdmaReduce(void *dst, const void *src, u64 cnt, const HcclDataType &datatype,
    const HcclReduceOp &redOp, Stream &stream)
{
    FftsSdmaSqeHeader sdmaSqeHeader;
    CHK_RET(GetSdmaSqeHeader(datatype, redOp, sdmaSqeHeader));
    CHK_RET(RefreshFftsDescSdma(dst, src, cnt * SIZE_TABLE[datatype], stream, sdmaSqeHeader.word));
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InitFftsSuccessorStart(HcclRtNotify signal, Stream &stream)
{
    u32 notifyID = 0;
    CHK_RET(hrtGetNotifyID(signal, &notifyID));

    u32 lastThreadPreIndex = fftsCtxsPtr->lastThreadIndex[stream.id()];
    if (lastThreadPreIndex == 0) {
        // 该stream首个task为signal record时，则忽略该signal
        fftsCtxsPtr->ignoredSuccessorStart.push_back(notifyID);
    } else {
        if (fftsCtxsPtr->unrelatedSuccessorEnd.find(notifyID) != fftsCtxsPtr->unrelatedSuccessorEnd.end()) {
            u32 successorIndex = fftsCtxsPtr->unrelatedSuccessorEnd[notifyID];
            fftsCtxsPtr->contexts[lastThreadPreIndex - 1]
                .successorList[fftsCtxsPtr->contexts[lastThreadPreIndex - 1].successorNum] = successorIndex;
            fftsCtxsPtr->contexts[lastThreadPreIndex - 1].successorNum++;
            fftsCtxsPtr->contexts[successorIndex].predCntInit++;
            fftsCtxsPtr->contexts[successorIndex].predCnt++;

            fftsCtxsPtr->unrelatedSuccessorEnd.erase(notifyID);
        } else {
            fftsCtxsPtr->unrelatedSuccessorStart[notifyID] = fftsCtxsPtr->lastThreadIndex[stream.id()] - 1;
        }
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InitFftsSuccessorEnd(HcclRtNotify signal, Stream &stream)
{
    u32 notifyID = 0;
    CHK_RET(hrtGetNotifyID(signal, &notifyID));

    if (fftsCtxsPtr->unassignedSuccessorEnd.find(stream.id()) == fftsCtxsPtr->unassignedSuccessorEnd.end()) {
        std::vector<u32> tmp;
        fftsCtxsPtr->unassignedSuccessorEnd.emplace(stream.id(), tmp);
    }
    fftsCtxsPtr->unassignedSuccessorEnd[stream.id()].push_back(notifyID);
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::SignalTaskParaSave(HcclRtNotify signal, Stream &stream, u32 userRank, u32 remoteUserRank,
    u64 offset, s32 stage, TaskType taskType, uint64_t beginTime)
{
    if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
        if (offset == INVALID_U64) {
            CHK_RET(hrtNotifyGetOffset(static_cast<HcclRtNotify>(signal), offset));
        }
        u64 notifyID = userRank;
        notifyID = (notifyID << 32) | (offset & 0x00000000FFFFFFFF);  // 0x00000000FFFFFFFF用于取offset的低32位
        remoteUserRank = (remoteUserRank == INVALID_UINT) ? static_cast<u32>(notifyID >> 32) : remoteUserRank;
        // 调用回调来保存task信息
        hccl::TaskParaNotify para(notifyID, stage, remoteUserRank, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.notify = para;
        taskPara.type = taskType;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_PROFILING;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
        GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
        if (offset == INVALID_U64) {
            CHK_RET(hrtNotifyGetOffset(static_cast<HcclRtNotify>(signal), offset));
        }
        u64 notifyID = userRank;
        notifyID = (notifyID << 32) | (offset & 0x00000000FFFFFFFF);  // 0x00000000FFFFFFFF用于取offset的低32位
        remoteUserRank = (remoteUserRank == INVALID_UINT) ? static_cast<u32>(notifyID >> 32) : remoteUserRank;
        hccl::TaskParaNotify para(notifyID, stage, remoteUserRank, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.notify = para;
        taskPara.type = taskType;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }

    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::SignalRecord(HcclRtNotify signal, Stream &stream, u32 userRank, u64 offset, s32 stage,
    bool inchip, u64 signalAddr, u32 notifyId)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    (void)notifyId;
    if (UNLIKELY(disableFfts_)) {
        return DispatcherPub::SignalRecord(signal, stream, userRank, offset, stage, inchip);
    }
    if (UNLIKELY(!FftsCtxReady())) {
        if (inchip) {
            CHK_RET(InitFftsSuccessorStart(signal, stream));
        } else {
            CHK_RET(InitFftsDescNotifyRecordRemote(signal, stream, signalAddr));
            CHK_RET(SignalTaskParaSave(signal, stream, userRank, INVALID_UINT,
                offset, stage, TaskType::TASK_NOTIFY_RECORD, beginTime));
        }
    } else {
        if (!inchip) {
            SkipFftsRefresh();
            CHK_RET(SignalTaskParaSave(signal, stream, userRank, INVALID_UINT,
                offset, stage, TaskType::TASK_NOTIFY_RECORD, beginTime));
        } else {
            // 卡内notify使用successor间依赖关系替代
        }
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::SignalWait(HcclRtNotify signal, Stream &stream, u32 userRank, u32 remoteUserRank, s32 stage,
    bool inchip, u32 notifyId, u32 timeOut)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    if (UNLIKELY(disableFfts_)) {
        return DispatcherPub::SignalWait(signal, stream, userRank, remoteUserRank, stage, inchip, notifyId, timeOut);
    }
    if (UNLIKELY(!FftsCtxReady())) {
        if (inchip) {
            CHK_RET(InitFftsSuccessorEnd(signal, stream));
        } else {
            CHK_RET(InitFftsDescNotifyWait(signal, stream));
            CHK_RET(SignalTaskParaSave(signal, stream, userRank, remoteUserRank,
                INVALID_U64, stage, TaskType::TASK_NOTIFY_WAIT, beginTime));
        }
    } else {
        if (!inchip) {
            SkipFftsRefresh();
            CHK_RET(SignalTaskParaSave(signal, stream, userRank, remoteUserRank,
                INVALID_U64, stage, TaskType::TASK_NOTIFY_WAIT, beginTime));
        } else {
            // 卡内notify使用successor间依赖关系替代
        }
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::MemcpyAsync(hccl::DeviceMem &dst, const hccl::DeviceMem &src, hccl::Stream &stream,
    u32 remoteUserRank, hccl::LinkType inLinkType)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    if (dst.size() < src.size()) {
        HCCL_ERROR(
            "The size of dst is smaller than that of src. dst addr[%p], dst size[%llu], src addr[%p], src size[%llu]",
            dst.ptr(), dst.size(), src.ptr(), src.size());
        return HCCL_E_PTR;
    }

    if (UNLIKELY(disableFfts_)) {
        return DispatcherPub::MemcpyAsync(dst, src, stream, remoteUserRank, inLinkType);
    }
    HCCL_INFO("Call FFTS MemcpyAsync. para: dst[%p] destMax[%llu] src[%p] count[%llu] "\
        "rtMemcpyKind[%d], inLinkType[%d]", dst.ptr(), dst.size(), src.ptr(), src.size(),
        HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, inLinkType);
    if (UNLIKELY(!FftsCtxReady())) {
        CHK_RET(InitFftsDescMemcpy(dst, src, stream));
    } else {
        CHK_RET(RefreshFftsDescMemcpy(dst, src, stream));
    }
    // 调用回调来保存task信息
    if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
        hccl::TaskParaDMA para(src.ptr(), dst.ptr(), src.size(), inLinkType, remoteUserRank,
            hccl::RdmaType::RDMA_TYPE_RESERVED, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.dma = para;
        taskPara.type = TaskType::TASK_SDMA;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_PROFILING;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
        GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
        hccl::TaskParaDMA para(src.ptr(), dst.ptr(), src.size(), inLinkType, remoteUserRank,
            hccl::RdmaType::RDMA_TYPE_RESERVED, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.dma = para;
        taskPara.type = TaskType::TASK_SDMA;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::ReduceAsync(const void *src, void *dst, u64 dataCount, const HcclDataType datatype,
    HcclReduceOp redOp, Stream &stream, HcclReduceType reduceType)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    if (UNLIKELY(disableFfts_)) {
        return DispatcherPub::ReduceAsync(src, dst, dataCount, datatype, redOp, stream, reduceType);
    }
    HCCL_INFO("Call FFTS ReduceAsync. para: src[%p] dst[%p] dataCount[%llu] datatype[%s] "\
        "redOp[%s], reduceType[%d].", src, dst, dataCount, GetDataTypeEnumStr(datatype).c_str(),
        GetReduceOpEnumStr(redOp).c_str(), reduceType);

    if (reduceType == HcclReduceType::HCCL_TBE_REDUCE) {
        // dtype=int64 或者 redOp=prod 都会走TbeReduce, 算法层控制的
        return TbeReduceAsync(src, dst, dataCount, datatype, redOp, stream, dst);
    }

    if (UNLIKELY(!FftsCtxReady())) {
        CHK_RET(InitFftsDescSdmaReduce(dst, src, dataCount, datatype, redOp, stream));
    } else {
        CHK_RET(RefreshFftsDescSdmaReduce(dst, src, dataCount, datatype, redOp, stream));
    }
    // 调用回调来保存 task 信息
    if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
        hccl::TaskParaReduce para(src, dst, dataCount * SIZE_TABLE[datatype], redOp, datatype,
            LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.reduce = para;
        taskPara.type = TaskType::TASK_REDUCE_INLINE;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_PROFILING;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
        GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
        hccl::TaskParaReduce para(src, dst, dataCount * SIZE_TABLE[datatype], redOp, datatype,
                LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.reduce = para;
        taskPara.type = TaskType::TASK_REDUCE_INLINE;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }

    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InlineReduceAsync(const void *src, u64 dataCount, const HcclDataType datatype,
    HcclReduceOp redOp, Stream &stream, void *dst, u32 remoteUserRank, hccl::LinkType inLinkType)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    if (UNLIKELY(disableFfts_)) {
        return DispatcherPub::InlineReduceAsync(
            src, dataCount, datatype, redOp, stream, dst, remoteUserRank, inLinkType);
    }
    HCCL_INFO("Call FFTS InlineReduceAsync. para: src[%p] dst[%p] dataCount[%llu] datatype[%s] "\
        "redOp[%s] inLinkType[%d] remoteUserRank[%u].", src, dst, dataCount, GetDataTypeEnumStr(datatype).c_str(),
        GetReduceOpEnumStr(redOp).c_str(), inLinkType, remoteUserRank);
    if (UNLIKELY(!FftsCtxReady())) {
        CHK_RET(InitFftsDescSdmaReduce(dst, src, dataCount, datatype, redOp, stream));
    } else {
        CHK_RET(RefreshFftsDescSdmaReduce(dst, src, dataCount, datatype, redOp, stream));
    }

    // 调用回调来保存 task 信息
    if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
        hccl::TaskParaReduce para(src, dst, dataCount * SIZE_TABLE[datatype], redOp, datatype, inLinkType,
            remoteUserRank, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.reduce = para;
        taskPara.type = TaskType::TASK_REDUCE_INLINE;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_PROFILING;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
        GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
        hccl::TaskParaReduce para(src, dst, dataCount * SIZE_TABLE[datatype], redOp, datatype, inLinkType,
            remoteUserRank, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.reduce = para;
        taskPara.type = TaskType::TASK_REDUCE_INLINE;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::RdmaSend(u32 dbindex, u64 dbinfo, const struct send_wr &wr, hccl::Stream &stream,
    u32 remoteUserRank)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    if (UNLIKELY(disableFfts_)) {
        return DispatcherPub::RdmaSend(dbindex, dbinfo, wr, stream, remoteUserRank);
    }
    // 调用回调来保存task信息
    if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
        // 0x00000000FFFFFFFF usrrank位于notifyID的高32位
        u64 notifyID = (static_cast<u64>(remoteUserRank) << 32) | (0x00000000FFFFFFFF);
        hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.buf_list[0].addr)),
                            reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dst_addr)),
                            wr.buf_list[0].len, notifyID, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_SEND_PAYLOAD,
                            fftsCtxsPtr->refreshIndex);
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.dma = para;
        taskPara.type = TaskType::TASK_RDMA;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_PROFILING;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
        GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
        u64 notifyID = (static_cast<u64>(remoteUserRank) << 32) | (0x00000000FFFFFFFF);
        hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.buf_list[0].addr)),
                            reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dst_addr)),
                            wr.buf_list[0].len, notifyID, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_SEND_PAYLOAD,
                            fftsCtxsPtr->refreshIndex);
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.dma = para;
        taskPara.type = TaskType::TASK_RDMA;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    return RdmaSendInternal(dbindex, dbinfo, stream);
}

HcclResult DispatcherFFTS::RdmaSend(u32 dbindex, u64 dbinfo, const struct send_wr &wr, hccl::Stream &stream,
    u32 userRank, u64 offset)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    if (UNLIKELY(disableFfts_)) {
        return DispatcherPub::RdmaSend(dbindex, dbinfo, wr, stream, userRank, offset);
    }

    // 调用回调来保存task信息
     if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
        // 0x00000000FFFFFFFF usrrank位于notifyID的高32位
        u64 notifyID = (static_cast<u64>(userRank) << 32) | (offset & 0x00000000FFFFFFFF);
        hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.buf_list[0].addr)),
                            reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dst_addr)),
                            wr.buf_list[0].len, notifyID, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_SEND_NOTIFY,
                            fftsCtxsPtr->refreshIndex);
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.dma = para;
        taskPara.type = TaskType::TASK_RDMA;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_PROFILING;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
        GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
        u64 notifyID = (static_cast<u64>(userRank) << 32) | (offset & 0x00000000FFFFFFFF);
        hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.buf_list[0].addr)),
                            reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dst_addr)),
                            wr.buf_list[0].len, notifyID, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_SEND_NOTIFY,
                            fftsCtxsPtr->refreshIndex);
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.dma = para;
        taskPara.type = TaskType::TASK_RDMA;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    return RdmaSendInternal(dbindex, dbinfo, stream);
}

HcclResult DispatcherFFTS::RdmaSendInternal(u32 dbindex, u64 dbinfo, hccl::Stream &stream)
{
    if (UNLIKELY(!FftsCtxReady())) {
        CHK_RET(InitFftsDescRdmaSend(dbindex, dbinfo, stream));
    } else {
        CHK_RET(RefreshFftsDescRdmaSend(dbindex, dbinfo, stream));
    }
    return HCCL_SUCCESS;
}

bool DispatcherFFTS::IsInvalidRdmaParam(u32 dbindex, u64 dbinfo)
{
    return dbindex == INVALID_UINT && dbinfo == INVALID_U64;
}

HcclResult DispatcherFFTS::ConstructFftsAicAivCtx(int count, void *addrListDevMemPtr, void *funcAddr, uint32_t blockDim,
    rtFftsPlusAicAivCtx_t *ctx, bool isInit)
{
    if (isInit) {
        ctx->successorNum = 0;
        ctx->aten = 0;

        ctx->prefetchConfig = 0;
        ctx->predCntInit = 0;
        ctx->predCnt = 0;

        ctx->prefetchOnceBitmap = 0;
        ctx->res6 = 0;
        ctx->prefetchEnableBitmap = 0;

        ctx->threadId = 0;
        ctx->threadDim = 1;

        ctx->res12 = 0;
        ctx->res13 = 0;
    }

    ctx->contextType = (count == 0) ? RT_CTX_TYPE_LABEL : RT_CTX_TYPE_AIV;

    const u64 u64HighMask = 0xffffffff00000000;
    const u64 u64LowMask = 0x00000000ffffffff;
    const u32 shift = 32;

    ctx->nonTailBlockdim = blockDim;
    ctx->tailBlockdim = blockDim;

    ctx->taskParamPtrBaseL = reinterpret_cast<u64>(addrListDevMemPtr) & u64LowMask;
    ctx->taskParamPtrBaseH = (reinterpret_cast<u64>(addrListDevMemPtr) & u64HighMask) >> shift;

    ctx->nonTailTaskStartPcL = reinterpret_cast<u64>(funcAddr) & u64LowMask;
    ctx->nonTailTaskStartPcH = (reinterpret_cast<u64>(funcAddr) & u64HighMask) >> shift;

    ctx->tailTaskStartPcL = reinterpret_cast<u64>(funcAddr) & u64LowMask;
    ctx->tailTaskStartPcH = (reinterpret_cast<u64>(funcAddr) & u64HighMask) >> shift;

    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::RefreshFftsDescVectorReduce(int count, void *addrListDevMemPtr, void *funcAddr,
    uint32_t blockDim, Stream &stream)
{
    rtFftsPlusComCtx_t &comCtx = fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex];
    rtFftsPlusAicAivCtx_t *ctx = reinterpret_cast<rtFftsPlusAicAivCtx_t *>(&comCtx);

    if (ctx->contextType != RT_CTX_TYPE_AIV && ctx->contextType != RT_CTX_TYPE_LABEL) {
        HCCL_ERROR("ffts context type is invaild, expected:0x%x, actual:0x%x", RT_CTX_TYPE_AIV, ctx->contextType);
        return HCCL_E_PARA;
    }

    ConstructFftsAicAivCtx(count, addrListDevMemPtr, funcAddr, blockDim, ctx, false);

    fftsCtxsPtr->refreshIndex++;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::InitFftsDescVectorReduce(int count, void *addrListDevMemPtr, void *funcAddr,
    uint32_t blockDim, Stream &stream)
{
    EnsureFftsContextsSize();
    rtFftsPlusComCtx_t &comCtx = fftsCtxsPtr->contexts[fftsCtxsPtr->refreshIndex];
    memset_s(&comCtx, sizeof(rtFftsPlusComCtx_t), 0, sizeof(rtFftsPlusComCtx_t));

    rtFftsPlusAicAivCtx_t *ctx = reinterpret_cast<rtFftsPlusAicAivCtx_t *>(&comCtx);

    ConstructFftsAicAivCtx(count, addrListDevMemPtr, funcAddr, blockDim, ctx, true);

    CHK_RET(UpdataFftsDescRelationship(stream));

    fftsCtxsPtr->refreshIndex++;
    fftsCtxsPtr->lastThreadIndex[stream.id()] = fftsCtxsPtr->refreshIndex;
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::VectorReduce(const void *src1, const void *src2, u64 count, const HcclDataType dataType,
    HcclReduceOp redOp, Stream &stream, const void *dst)
{
    void *funcAddr = nullptr;
    uint32_t blockDim = 0;
    void *addrListDevMemPtr = nullptr;
    void *argsHandle = nullptr;
    if (count != 0) {
#ifndef HCCD
        CHK_RET(tbeVectorReduce_->PrepareVectorReduce(src1, src2, count, dataType, redOp, dst, stream.ptr(),
            addrListDevMemPtr, argsHandle, funcAddr, blockDim));
#else
        HCCL_ERROR("[DispatcherFFTS][VectorReduce] does not support this interface.");
        return HCCL_E_PARA;
#endif
        argsHandleList.push_back(argsHandle);
    }
    CHK_RET(SetFftsDescVectorReduce(src1, dst, count, addrListDevMemPtr, funcAddr, blockDim, dataType, redOp, stream));
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::VectorReduceLoop(const void *src1, const void *src2, u64 count, const HcclDataType dataType,
    HcclReduceOp redOp, Stream &stream, const void *dst)
{
    const u32 unitSize = SIZE_TABLE[dataType];
    void *currentSrc1 = const_cast<void *>(src1);
    void *currentSrc2 = const_cast<void *>(src2);
    void *currentDst = const_cast<void *>(dst);
 
    // 计算出字节数为32字节整倍数的最大count
    const u64 maxCountPerLoop = ((TBE_REDUCE_MAX_COUNT * unitSize) / UB_BLOCK_SIZE * UB_BLOCK_SIZE) / unitSize;
 
    u64 countLeft = count;

    // 使用do while循环，是为了保证count为0时也进入一次VectorReduce，避免子图复用出错
    do  {
        u64 currentCount = countLeft > maxCountPerLoop ? maxCountPerLoop : countLeft;
        HCCL_DEBUG(
            "[VectorReduceLoop] currentCount[%llu], countLeft[%llu], currentSrc1[%p], currentSrc2[%p], currentDst[%p]",
            currentCount, countLeft, currentSrc1, currentSrc2, currentDst);
 
        CHK_RET(VectorReduce(currentSrc1, currentSrc2, currentCount, dataType, redOp, stream, currentDst));
 
        currentSrc1 = static_cast<void *>(static_cast<s8 *>(currentSrc1) + currentCount * unitSize);
        currentSrc2 = static_cast<void *>(static_cast<s8 *>(currentSrc2) + currentCount * unitSize);
        currentDst = static_cast<void *>(static_cast<s8 *>(currentDst) + currentCount * unitSize);
        countLeft -= currentCount;
    } while (countLeft > 0);

    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::SetFftsTailVectorReduceDescSdma(void *devMem, const void *tailSrc, u64 count,
    const HcclDataType dataType, HcclReduceOp redOp, Stream &stream)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    if (UNLIKELY(!FftsCtxReady())) {
        CHK_RET(InitFftsDescSdma(devMem, tailSrc, count, stream, SDMA_FP32_ATOMIC_MOVE_SQE));
    } else {
        CHK_RET(RefreshFftsDescSdma(devMem, tailSrc, count, stream, SDMA_FP32_ATOMIC_MOVE_SQE));
    }

    // 调用回调来保存 task 信息
    if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
        hccl::TaskParaDMA para(tailSrc, devMem, count, LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID,
            hccl::RdmaType::RDMA_TYPE_RESERVED, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.dma = para;
        taskPara.type = TaskType::TASK_SDMA;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_PROFILING;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
        GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
        hccl::TaskParaDMA para(tailSrc, devMem, count, LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID,
            hccl::RdmaType::RDMA_TYPE_RESERVED, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.dma = para;
        taskPara.type = TaskType::TASK_SDMA;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::SetFftsDescVectorReduce(const void *src, const void *dst, int count, void *addrListDevMemPtr,
    void *funcAddr, uint32_t blockDim, const HcclDataType dataType, HcclReduceOp redOp, Stream &stream)
{
    uint64_t beginTime = GetMsprofSysCycleTime();
    if (UNLIKELY(!FftsCtxReady())) {
        CHK_RET(InitFftsDescVectorReduce(count, addrListDevMemPtr, funcAddr, blockDim, stream));
    } else {
        CHK_RET(RefreshFftsDescVectorReduce(count, addrListDevMemPtr, funcAddr, blockDim, stream));
    }

    // 调用回调来保存 task 信息
    if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
        hccl::TaskParaReduce para(src, dst, count * SIZE_TABLE[dataType], redOp, dataType,
            LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.reduce = para;
        taskPara.type = TaskType::TASK_REDUCE_TBE;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_PROFILING;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
        GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
        hccl::TaskParaReduce para(src, dst, count * SIZE_TABLE[dataType], redOp, dataType,
            LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID, (fftsCtxsPtr->refreshIndex - 1));
        struct TaskPara taskPara;
        taskPara.stream = stream.ptr();
        taskPara.isMainStream = stream.IsMainStream();
        taskPara.beginTime = beginTime;
        taskPara.reduce = para;
        taskPara.type = TaskType::TASK_REDUCE_TBE;
        taskPara.isFftsDispatcher = true;
        taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
        callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
    }
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::TailVectorReduce(const void *tailSrc1, const void *tailSrc2, u64 tailCount,
    const HcclDataType dataType, HcclReduceOp redOp, Stream &stream, void *tailDst)
{
    void *devMem1 = nullptr;
    void *devMem2 = nullptr;
    void *addrListDevMemPtr = nullptr;
    void *funcAddr = nullptr;
    void *argsHandle = nullptr;
    uint32_t blockDim = 0;
    if (tailCount != 0) {
#ifndef HCCD
        CHK_RET(tbeVectorReduce_->PrepareTailVectorReduce(devMem1, devMem2, tailCount, dataType, redOp, stream.ptr(),
            addrListDevMemPtr, argsHandle, funcAddr, blockDim));
#else
        HCCL_ERROR("[DispatcherFFTS][VectorReduce] does not support this interface.");
        return HCCL_E_PARA;
#endif
        argsHandleList.push_back(argsHandle);
    } else {
#ifndef HCCD
        // tailCount=0也要判断不支持某设备类型+数据类型+reduce类型，直接返回退出报错
        CHK_RET(tbeVectorReduce_->JudgeIsSupported(dataType, redOp));
#endif
    }
    u64 dataCount = tailCount * SIZE_TABLE[dataType];

    CHK_RET(SetFftsTailVectorReduceDescSdma(devMem1, tailSrc1, dataCount, dataType, redOp, stream));
    CHK_RET(SetFftsTailVectorReduceDescSdma(devMem2, tailSrc2, dataCount, dataType, redOp, stream));
    CHK_RET(SetFftsDescVectorReduce(devMem1, tailDst, tailCount, addrListDevMemPtr, funcAddr, blockDim, dataType,
        redOp, stream));
    CHK_RET(SetFftsTailVectorReduceDescSdma(tailDst, devMem2, dataCount, dataType, redOp, stream));
    return HCCL_SUCCESS;
}

HcclResult DispatcherFFTS::TbeReduceAsync(const void *src1, const void *src2, u64 count, const HcclDataType dataType,
    HcclReduceOp redOp, Stream &stream, const void *dst)
{
    void *tailSrc1 = nullptr;
    void *tailSrc2 = nullptr;
    void *tailDst = nullptr;
    u64 headCount = 0;
    u64 tailCount = 0;
#ifndef HCCD
    CHK_RET(tbeVectorReduce_->JudgeIsTail(src1, src2, dst, count, dataType, headCount, tailCount, tailSrc1, tailSrc2,
        tailDst));
#else
        HCCL_ERROR("[DispatcherFFTS][TbeReduceAsync] does not support this interface.");
        return HCCL_E_PARA;
#endif
    CHK_RET(VectorReduceLoop(src1, src2, headCount, dataType, redOp, stream, dst));
    CHK_RET(TailVectorReduce(tailSrc1, tailSrc2, tailCount, dataType, redOp, stream, tailDst));

    return HCCL_SUCCESS;
}

} // namespace hccl
